package com.xc.quickstart.infrastructure.mq;

import com.alibaba.fastjson.JSON;
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.PropertyValueConst;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.xc.quickstart.common.properties.MqProperties;
import com.xc.quickstart.mq.listener.BaseMqListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Properties;
import java.util.Set;
import java.util.stream.Collectors;

@Slf4j
@ConditionalOnWebApplication
public class MqListenerConfig {

    @Autowired
    private MqProperties mqProperties;

    @Autowired
    private List<BaseMqListener> baseMessageListenerList;

    @PostConstruct
    public void init() {
        baseMessageListenerList.forEach(listener -> {
            Properties properties = new Properties();
            properties.put(PropertyKeyConst.AccessKey, mqProperties.getAccessKey());
            properties.put(PropertyKeyConst.SecretKey, mqProperties.getSecretKey());
            properties.put(PropertyKeyConst.NAMESRV_ADDR, mqProperties.getNameSrvAddr());
            properties.put(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
            properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
            properties.put(PropertyKeyConst.ConsumeThreadNums, listener.getConsumeThreadNums());
            properties.put(PropertyKeyConst.GROUP_ID, listener.getGroupId());
            Consumer consumer = ONSFactory.createConsumer(properties);
            consumer.subscribe(listener.getTopic(),getTags(listener.getTags()), listener);
            consumer.start();

            log.info("消息监听注册成功,{}", JSON.toJSONString(listener));
        });
    }

    private String getTags(Set<String> tagSet) {
        if (CollectionUtils.isEmpty(tagSet)) {
            return "*";
        }

        return tagSet.stream().filter(StringUtils::isNotBlank).collect(Collectors.joining("||"));
    }
}
